package org.swqsbssddqs.shortlink.project.mq.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import org.swqsbssddqs.shortlink.project.common.convention.exception.ServiceException;
import org.swqsbssddqs.shortlink.project.dao.entity.*;
import org.swqsbssddqs.shortlink.project.dao.mapper.*;
import org.swqsbssddqs.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import org.swqsbssddqs.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import org.swqsbssddqs.shortlink.project.mq.producer.DelayShortLinkStatsProducer;

import java.time.LocalDate;
import java.time.LocalTime;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import static org.swqsbssddqs.shortlink.project.common.constant.RedisCacheConstant.LOCK_UPDATE_GID_KEY;
@Component
@RequiredArgsConstructor
@Slf4j
public class ShortLinkStatsSaveConsumer  implements StreamListener<String, MapRecord<String, String, String>> {

    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final RedissonClient redissonClient;
    private final ShortLinkAccessStatsMapper shortLinkAccessStatsMapper;
    private final ShortLinkLocaleStatsMapper shortLinkLocaleStatsMapper;
    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkOsStatsMapper shortLinkOsStatsMapper;
    private final ShortLinkBrowserStatsMapper shortLinkBrowserStatsMapper;
    private final ShortLinkDeviceStatsMapper shortLinkDeviceStatsMapper;
    private final ShortLinkNetworkStatsMapper shortLinkNetworkStatsMapper;
    private final ShortLinkAccessLogsMapper shortLinkAccessLogsMapper;
    private final ShortLinkStatsTodayMapper linkStatsTodayMapper;
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    @Value("${short-link.locale.amap-api-key}")
    private String amap_api_key;
    @Value("${short-link.locale.url}")
    private String amap_api_url;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        if(messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())){
            if(messageQueueIdempotentHandler.isAccomplish(id.toString())){
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            Map<String, String> messageValue = message.getValue();
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(messageValue.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            actualShortLinkStats(statsRecord);
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
//            throw new RuntimeException();
        } catch (Throwable ex) {
            // 某某某情况宕机了
            log.error("记录短链接监控消费异常", ex);
            messageQueueIdempotentHandler.delMessage(id.toString());
            throw ex;
        }
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }
    public void actualShortLinkStats(ShortLinkStatsRecordDTO param){
        String fullShortUrl = param.getFullShortUrl();
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_UPDATE_GID_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        if(!rLock.tryLock()){
            log.error("获取读锁失败，加入阻塞队列");
            delayShortLinkStatsProducer.send(param);
            return;
        }
        try {
            log.info("获取读锁成功，更新stats");
            Date currentDate = param.getCurrentDate();
            //统计ipV//统计uv，pv
            String actualIp = param.getRemoteAddr();
            ShortLinkAccessStatsDO accessStatsDO = ShortLinkAccessStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .uv(param.getUvFirstFlag()?1:0)
                    .pv(1)
                    .uip(param.getUipFirstFlag()?1:0)
                    .hour(LocalTime.now().getHour())
                    .weekday(LocalDate.now().getDayOfWeek().getValue())
                    .build();
            shortLinkAccessStatsMapper.shortLinkAccessStats(accessStatsDO);


            //统计地区相关
            HashMap<String, Object> map = new HashMap<>();
            map.put("key",amap_api_key);
            map.put("ip",actualIp);
            String s = HttpUtil.get(amap_api_url,map);
            JSONObject localeResultObj = JSON.parseObject(s);
            String infoCode = localeResultObj.getString("infocode");
            String actualProvince = "未知";
            String actualCity = "未知";
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                String province = localeResultObj.getString("province");
                boolean unknownFlag = StrUtil.equals(province, "[]");
                ShortLinkLocaleStatsDO linkLocaleStatsDO = ShortLinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .country("中国")
                        .date(currentDate)
                        .build();
                shortLinkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
            }

            //统计操作系统相关
            String os = param.getOs();
            ShortLinkOsStatsDO linkOsStatsDO = ShortLinkOsStatsDO.builder()
                    .os(os)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            shortLinkOsStatsMapper.shortLinkOsState(linkOsStatsDO);

            //统计浏览器相关
            String browser = param.getBrowser();
            ShortLinkBrowserStatsDO browserStatsDO = ShortLinkBrowserStatsDO.builder()
                    .browser(browser)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            shortLinkBrowserStatsMapper.shortLinkBrowserState(browserStatsDO);

            //统计设备
            String device = param.getDevice();
            ShortLinkDeviceStatsDO deviceStatsDO = ShortLinkDeviceStatsDO.builder()
                    .device(device)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            shortLinkDeviceStatsMapper.shortLinkDeviceState(deviceStatsDO);

            //统计网络
            String network = param.getNetwork();
            ShortLinkNetworkStatsDO networkStatsDO = ShortLinkNetworkStatsDO.builder()
                    .network(network)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            shortLinkNetworkStatsMapper.shortLinkNetworkState(networkStatsDO);

            //为实现高频IP访问，记录访问日志
            ShortLinkAccessLogsDO accessLogsDO = ShortLinkAccessLogsDO.builder()
                    .browser(browser)
                    .user(param.getUv())
                    .network(network)
                    .locale(StrUtil.join("-","中国",actualProvince,actualCity))
                    .device(device)
                    .fullShortUrl(fullShortUrl)
                    .os(os)
                    .locale(StrUtil.join("-","中国",actualProvince,actualCity))
                    .ip(actualIp)
                    .build();
            shortLinkAccessLogsMapper.insert(accessLogsDO);
            LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
            String gid = shortLinkGotoDO.getGid();
            shortLinkMapper.uVpVipVIncrement(gid,fullShortUrl,1,param.getUvFirstFlag()?1:0,param.getUipFirstFlag()?1:0);
            ShortLinkStatsTodayDO linkStatsTodayDO = ShortLinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUip(param.getUvFirstFlag()? 1 : 0)
                    .todayUv(param.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } finally {
            rLock.unlock();
        }
    }
}
